package com.cloud.job.consumer;

import com.cloud.api.domain.job.VehicleData;
import com.cloud.job.doris.DorisStreamLoad;
import com.cloud.job.utils.DateUtils;
import com.cloud.job.utils.JsonUtils;
import com.fasterxml.jackson.databind.node.ObjectNode;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

import static com.cloud.job.utils.JsonUtils.*;

/**
 * @author 杨斌
 * 2024-12-2024-12-18
 */
@Slf4j
@Component
@ConfigurationProperties
public class MessageConsumer {
    @Resource
    private DorisStreamLoad dorisStreamLoad;

    @KafkaListener(id="${listener.consumer1.id}",topics = "#{'${listener.consumer1.topics}'.split(',')}", groupId = "${listener.consumer1.groupId}",concurrency = "${listener.consumer1.concurrency}")
    public void processMessage(@Payload List<ConsumerRecord<String, String>> consumerRecords,
                               Acknowledgment acknowledgment) {

        log.info("Starting to process real messages...");
        List<VehicleData> vehicleDataList = new ArrayList<>();
        try  {
            log.info("Polled {} real messages ",consumerRecords.size());
            if (consumerRecords.isEmpty()) {
                log.info("No real messages to process");
                return; // 返回空列表
            }
            for (ConsumerRecord<String, String> record : consumerRecords) {
                try {
                    VehicleData vehicleData = parseMessage(record.value());
                    vehicleDataList.add(vehicleData);
                } catch (Exception e) {
                    log.error("Error parsing real messages: {}", record.value(), e);
                }
            }
            if (!vehicleDataList.isEmpty()) {
                String jsonString = JsonUtils.writeValueAsString(vehicleDataList);
                long loadStartTime = System.currentTimeMillis();
                dorisStreamLoad.loadJson(jsonString, VehicleData.class);
                long loadEndTime = System.currentTimeMillis();
                log.info("Inserted real messages {} records into database in {} ms", vehicleDataList.size(), (loadEndTime - loadStartTime));
                // 提交偏移量
                acknowledgment.acknowledge();
                log.info("Committed offsets for processed real messages.");
            } else {
                log.info("No parsed real messages to insert into database");
            }
        } catch (Exception e) {
            log.error("Error real processing real messages", e);
        } finally {
            log.info("Finished real processing real messages.");
        }
    }

    private VehicleData parseMessage(String message) throws Exception {
        ObjectNode objectNode = JsonUtils.readObjectNode(message);
        VehicleData.VehicleDataBuilder vehicleDataBuilder = VehicleData.builder();

        String opTime = getTextValue(objectNode, "dataContent", "localDateTime");
        String dt = DateUtils.parseAndFormatToDt(opTime);

        vehicleDataBuilder
                .opTime(opTime)
                .vin(getTextValue(objectNode, "header", "vin"))
                .dt(dt)
                .commandType(getTextValue(objectNode, "header", "commandType"))
                .vehicleState(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "vehicleState"))
                .chargeState(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "chargeState"))
                .operationalMode(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "operationalMode"))
                .speed(getIntValue(objectNode, "dataContent", "evData", "vehicleData", "speed"))
                .accumulativeMileage(getLongValue(objectNode, "dataContent", "evData", "vehicleData", "accumulativeMileage"))
                .totalVoltage(getIntValue(objectNode, "dataContent", "evData", "vehicleData", "totalVoltage"))
                .totalCurrent(getIntValue(objectNode, "dataContent", "evData", "vehicleData", "totalCurrent"))
                .soc(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "soc"))
                .dcState(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "dcState"))
                .gear(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "gear"))
                .insulationResistance(getIntValue(objectNode, "dataContent", "evData", "vehicleData", "insulationResistance"))
                .accelerationPedalTravelValue(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "accelerationPedalTravelValue"))
                .brakePedalStatus(getShortValue(objectNode, "dataContent", "evData", "vehicleData", "brakePedalStatus"))
                .driveMotorCount(getShortValue(objectNode, "dataContent", "evData", "driveMotorCount"))
                .driveMotorData(getDriveMotorDataList(objectNode, "dataContent", "evData", "driveMotorDatas"))
                .fuelcellVoltage(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "voltage"))
                .fuelcellCurrent(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "current"))
                .fuelConsumption(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "fuelConsumption"))
                .temperatureProbeCount(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "temperatureProbeCount"))
                .probeTemperature(getProbeTemperatureList(objectNode, "dataContent", "evData", "fuelCellData", "probeTemperature"))
                .hydrogenSystemMaxTemperature(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "hydrogenSystemMaxTemperature"))
                .hydrogenSystemMaxTemperatureProbeNum(getShortValue(objectNode, "dataContent", "evData", "fuelCellData", "hydrogenSystemMaxTemperatureProbeNum"))
                .hydrogenSystemMaxConcentration(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "hydrogenSystemMaxConcentration"))
                .hydrogenSystemMaxConcentrationProbeNum(getShortValue(objectNode, "dataContent", "evData", "fuelCellData", "hydrogenSystemMaxConcentrationProbeNum"))
                .hydrogenSystemMaxPressure(getIntValue(objectNode, "dataContent", "evData", "fuelCellData", "hydrogenSystemMaxPressure"))
                .hydrogenSystemMaxPressureProbeNum(getShortValue(objectNode, "dataContent", "evData", "fuelCellData", "hydrogenSystemMaxPressureProbeNum"))
                .dcStatus(getBooleanValue(objectNode, "dataContent", "evData", "fuelCellData", "dcStatus"))
                .engineStatus(getShortValue(objectNode, "dataContent", "evData", "engineData", "Status"))
                .engineCrankshaftSpeed(getIntValue(objectNode, "dataContent", "evData", "engineData", "crankshaftSpeed"))
                .engineFuelConsumption(getShortValue(objectNode, "dataContent", "evData", "engineData", "fuelConsumption"))
                .lat(getLongValue(objectNode, "dataContent", "evData", "locationData", "latitude"))
                .latitudeDataType(getTextValue(objectNode, "dataContent", "evData", "locationData", "latitudeDataType"))
                .lng(getLongValue(objectNode, "dataContent", "evData", "locationData", "longitude"))
                .longitudeDataType(getTextValue(objectNode, "dataContent", "evData", "locationData", "longitudeDataType"))
                .valid(getBooleanValue(objectNode, "dataContent", "evData", "locationData", "valid"))
                .status(getShortValue(objectNode, "dataContent", "evData", "locationData", "status"))
                .maxVoltageSystemNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "maxVoltageSystemNum"))
                .maxVoltageBatteryNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "maxVoltageBatteryNum"))
                .batteryMaxVoltage(getIntValue(objectNode, "dataContent", "evData", "extremeData", "batteryMaxVoltage"))
                .minVoltageSystemNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "minVoltageSystemNum"))
                .minVoltageBatteryNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "minVoltageBatteryNum"))
                .batteryMinVoltage(getShortValue(objectNode, "dataContent", "evData", "extremeData", "batteryMinVoltage"))
                .maxTemperatureSystemNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "maxTemperatureSystemNum"))
                .maxTemperatureNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "maxTemperatureNum"))
                .maxTemperature(getShortValue(objectNode, "dataContent", "evData", "extremeData", "maxTemperature"))
                .minTemperatureSystemNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "minTemperatureSystemNum"))
                .minTemperatureNum(getShortValue(objectNode, "dataContent", "evData", "extremeData", "minTemperatureNum"))
                .minTemperature(getShortValue(objectNode, "dataContent", "evData", "extremeData", "minTemperature"))
                .level(getShortValue(objectNode, "dataContent", "evData", "alarmData", "level"))
                .alarmInfo(getLongValue(objectNode, "dataContent", "evData", "alarmData", "alarmInfo"))
                .temperatureDifference(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "temperatureDifference"))
                .batteryHighTemperature(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "batteryHighTemperature"))
                .deviceTypeOverVoltage(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "deviceTypeOverVoltage"))
                .deviceTypeUnderVoltage(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "deviceTypeUnderVoltage"))
                .socLow(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "socLow"))
                .monomerBatteryOverVoltage(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "monomerBatteryOverVoltage"))
                .monomerBatteryUnderVoltage(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "monomerBatteryUnderVoltage"))
                .socHigh(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "socHigh"))
                .socJump(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "socJump"))
                .deviceTypeMisMatch(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "deviceTypeMisMatch"))
                .batteryConsistencyPoor(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "batteryConsistencyPoor"))
                .insulation(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "insulation"))
                .dcTemperature(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "dcTemperature"))
                .brakingSystem(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "brakingSystem"))
                .driveMotorControllerTemperature(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "driveMotorControllerTemperature"))
                .highPressureInterlock(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "highPressureInterlock"))
                .driveMotorTemperature(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "driveMotorTemperature"))
                .deviceTypeOverFilling(getBooleanValue(objectNode, "dataContent", "evData", "alarmData", "deviceTypeOverFilling"))
                .deviceFailureCount(getShortValue(objectNode, "dataContent", "evData", "alarmData", "deviceFailureCount"))
                .deviceFailureCodes(getLongListValue(objectNode, "dataContent", "evData", "alarmData", "deviceFailureCodes"))
                .driveMotorFailureCount(getShortValue(objectNode, "dataContent", "evData", "alarmData", "driveMotorFailureCount"))
                .driveMotorFailureCodes(getLongListValue(objectNode, "dataContent", "evData", "alarmData", "driveMotorFailureCodes"))
                .engineFailureCount(getShortValue(objectNode, "dataContent", "evData", "alarmData", "engineFailureCount"))
                .engineFailureCodes(getLongListValue(objectNode, "dataContent", "evData", "alarmData", "engineFailureCodes"))
                .otherFailure(getShortValue(objectNode, "dataContent", "evData", "alarmData", "otherFailure"))
                .otherFailureCodes(getLongListValue(objectNode, "dataContent", "evData", "alarmData", "otherFailureCodes"))
                .subsystemVoltageCount(getShortValue(objectNode, "dataContent", "evData", "subsystemVoltageCount"))
                .subsystemVoltageDatas(getSubSystemVoltageDataList(objectNode, "dataContent", "evData", "subsystemVoltageDatas"))
                .subsystemTemperatureCount(getShortValue(objectNode, "dataContent", "evData", "subsystemTemperatureCount"))
                .subsystemTemperatures(getSubSystemTemperatureList(objectNode, "dataContent", "evData", "subsystemTemperatures"))
                .lockVehicleState(getShortValue(objectNode, "dataContent", "evData", "lockVehicleState", "lockVehicleState"))
                .supportLock(getIntValue(objectNode, "dataContent", "evData", "lockVehicleState", "supportLock"))
                .commandId(getTextValue(objectNode, "dataContent", "evData", "lockVehicleState", "commandId"))
                .startOffset(getStartOffsetMap(objectNode, "dataContent", "evData", "startOffset"))
                .receiveTime(getTextValue(objectNode, "receive_time"))
                .record(getTextValue(objectNode, "record"))
                .vehicleType(getTextValue(objectNode, "vehicle_type"))
                // 新增自定义属性
                .customDataLength(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "customDataLength"))
                .st86Version(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "st86Version"))
                .stm32Version(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "stm32Version"))
                .deviceVersion(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "deviceVersion"))
                .userCode(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "userCode"))
                .vehicleCode(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "vehicleCode"))
                .lockStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "lockStatus"))
                .canStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "canStatus"))
                .antennaStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "antennaStatus"))
                .powerOffStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "powerOffStatus"))
                .sdCardStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "sdCardStatus"))
                .drivingDirection(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "drivingDirection"))
                .lockEnableStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "lockEnableStatus"))
                .chargingCycleCount(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "chargingCycleCount"))
                .accumulatedChargingEnergy(getLongValue(objectNode, "dataContent", "evData", "customMessageData", "accumulatedChargingEnergy"))
                .accumulatedWorkingTime(getLongValue(objectNode, "dataContent", "evData", "customMessageData", "accumulatedWorkingTime"))
                .accumulatedGpsMileage(getLongValue(objectNode, "dataContent", "evData", "customMessageData", "accumulatedGpsMileage"))
                .totalStorageCapacity(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "totalStorageCapacity"))
                .remainingStorageCapacity(getIntValue(objectNode, "dataContent", "evData", "customMessageData", "remainingStorageCapacity"))
                .gpsSignalStrength(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "gpsSignalStrength"))
                .gsmSignalStrength(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "gsmSignalStrength"))
                .keyLockStatus(getShortValue(objectNode, "dataContent", "evData", "customMessageData", "keyLockStatus"))
                .keyLock(getTextValue(objectNode, "dataContent", "evData", "customMessageData", "key"));
        return vehicleDataBuilder.build();
    }

}
